package com.feiluan.mqtt.subscriber;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.feiluan.config.mqtt.*;
import com.feiluan.mqtt.publisher.*;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;


/**
 * MQTT Subscriber Class
 *
 * @author Moniruzzaman Md
 */

@Component
public class MQTTSubcriber extends MQTTSubcriberConfig implements MqttCallbackExtended, MQTTSubscriberBase {

    private String brokerUrl = null;
    final private String colon = ":";
    final private String clientId = UUID.randomUUID().toString();


    private MqttClient mqttClient = null;
    private MqttConnectOptions connectionOptions = null;
    private MemoryPersistence persistence = null;

    private static final Logger logger = LoggerFactory.getLogger(MQTTSubcriberConfig.class);

    public MQTTSubcriber() {
        this.config();
    }
//
//    @Autowired
//    MQTTPublisher publisher;
    @Autowired
    MQTTCloudBoxPublisher cloudBoxPublisher;
    @Autowired
    MQTTLampControlPublisher lampControlPublisher;
    @Autowired
    MQTTCameraPublisher cameraPublisher;
    @Autowired
    MQTTEnvironmentalSensingPublisher environmentalSensingPublisher;
    @Autowired
    MQTTPhotoelectricBoxPublisher photoelectricBoxPublisher;
    @Autowired
    MQTTInformationScreenPublisher informationScreenPublisher;
    @Autowired
    MQTTInteractiveScreenPublisher interactiveScreenPublisher;
    @Autowired
    MQTTCrowdEarlyWarningPublisher crowdEarlyWarningPublisher;
    @Autowired
    MQTTAntiTheftLockPublisher antiTheftLockPublisher;
    @Autowired
    MQTTChargingPilePublisher chargingPilePublisher;
    @Autowired
    MQTTManholeCoverPublisher manholeCoverPublisher;
    @Autowired
    MQTTOneButtonHelpPublisher oneButtonHelpPublisher;
    @Autowired
    MQTTRadioBroadcastPublisher radioBroadcastPublisher;
    @Autowired
    MQTTSmokeSensorPublisher smokeSensorPublisher;
    @Autowired
    MQTTTrashPublisher trashPublisher;
    @Autowired
    MQTTWifiPublisher wifiPublisher;
    @Autowired
    MQTTWaterImmersionPublisher waterImmersionPublisher;


    /*
     * (non-Javadoc)
     *
     * @see org.eclipse.paho.client.mqttv3.MqttCallback#connectionLost(java.lang.
     * Throwable)
     */
    @Override
    public void connectionLost(Throwable cause) {

        logger.info("Connection Lost");

    }

    /*
     * (non-Javadoc)
     *
     * @see
     * org.eclipse.paho.client.mqttv3.MqttCallback#messageArrived(java.lang.String,
     * org.eclipse.paho.client.mqttv3.MqttMessage)
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {

        // 接受到的消息 ，字符串转换为Json 对象
        JSONObject jsonObject = JSONObject.parseObject(new String(message.getPayload()));

        //将原来json中的update 节点， 从unix time  转换为  yyyy-MM-dd HH:mm:ss 的时间格式
        String date = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new java.util.Date(Long.parseLong(jsonObject.getString("updateTime"))));

        // 更新update
        jsonObject.put("updateTime", date);

        //获取devType字段
        String object = jsonObject.getString("devType");

        //封装发送的message
        Map params = new HashMap();
        params.put("params", jsonObject);


        switchAndProccess(object, JSON.toJSONString(params));
    }


    private void switchAndProccess(String object, String message) {

        // 云盒
        if ("0".equals(object)) {
            cloudBoxPublisher.publishMessage(MQTTCloudBoxPublishConfig.topic, message);
        }
        //灯控
        if ("1".equals(object)) {
            lampControlPublisher.publishMessage(MQTTLampControlPublishConfig.topic, message);
        }
        //摄像头
        if ("2".equals(object)) {
            cameraPublisher.publishMessage(MQTTCameraPublishConfig.topic, message);
        }
        //防盗锁
        if ("3".equals(object)) {
            antiTheftLockPublisher.publishMessage(MQTTAntiTheftLockPublishConfig.topic, message);
        }

        // Wifi
        if ("4".equals(object)) {
            wifiPublisher.publishMessage(MQTTWifiPublishConfig.topic, message);
        }
        //信息屏
        if ("5".equals(object)) {
            informationScreenPublisher.publishMessage(MQTTInformationScreenPublishConfig.topic, message);
        }

        //光电盒
        if ("7".equals(object)) {
            photoelectricBoxPublisher.publishMessage(MQTTPhotoelectricBoxPublishConfig.topic, message);
        }
        //环境传感
        if ("8".equals(object)) {
            environmentalSensingPublisher.publishMessage(MQTTEnvironmentalSensingPublishConfig.topic, message);
        }
        // 交互屏
        if ("11".equals(object)) {
            interactiveScreenPublisher.publishMessage(MQTTInteractiveScreenPublishConfig.topic, message);
        }
        // 人群预警
        if ("12".equals(object)) {
            crowdEarlyWarningPublisher.publishMessage(MQTTCrowdEarlyWarningPublishConfig.topic, message);
        }
        // 井盖
        if ("16".equals(object)) {
            manholeCoverPublisher.publishMessage(MQTTManholeCoverPublishConfig.topic, message);
        }

        // 垃圾桶
        if ("17".equals(object)) {
            trashPublisher.publishMessage(MQTTTrashPublishConfig.topic, message);
        }
        //广播
        if ("18".equals(object)) {
            radioBroadcastPublisher.publishMessage(MQTTRadioBroadcastPublishConfig.topic, message);
        }
        //一键求助
        if ("29".equals(object)) {
            oneButtonHelpPublisher.publishMessage(MQTTOneButtonHelpPublishConfig.topic, message);
        }
        // 充电桩
        if ("30".equals(object)) {
            chargingPilePublisher.publishMessage(MQTTChargingPilePublishConfig.topic, message);
        }
        // 水浸
        if ("32".equals(object)) {
            waterImmersionPublisher.publishMessage(MQTTWaterImmersionPublishConfig.topic, message);
        }
        // 烟雾传感器
        if ("33".equals(object)) {
            smokeSensorPublisher.publishMessage(MQTTSmokeSensorPublishConfig.topic, message);
        }










       /* //实例
        else if ("19".equals(object)) {
            publisher.publishMessage(MQTTPublishConfig.topic, message);
        }*/
    }

    /*
     * (non-Javadoc)
     *
     * @see
     * org.eclipse.paho.client.mqttv3.MqttCallback#deliveryComplete(org.eclipse.paho
     * .client.mqttv3.IMqttDeliveryToken)
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // Leave it blank for subscriber

    }

    /*
     * (non-Javadoc)
     *
     * @see
     * com.monirthought.mqtt.subscriber.MQTTSubscriberBase#subscribeMessage(java.
     * lang.String)
     */
    @Override
    public void subscribeMessage(String topic) {
        try {
            this.mqttClient.subscribe(topic, this.qos);
        } catch (MqttException me) {
            me.printStackTrace();
        }
    }

    /*
     * (non-Javadoc)
     *
     * @see com.monirthought.mqtt.subscriber.MQTTSubscriberBase#disconnect()
     */
    @Override
    public void disconnect() {
        try {
            this.mqttClient.disconnect();
        } catch (MqttException me) {
            logger.error("ERROR", me);
        }
    }

    /*
     * (non-Javadoc)
     *
     * @see com.monirthought.config.MQTTConfig#config(java.lang.String,
     * java.lang.Integer, java.lang.Boolean, java.lang.Boolean)
     */
    @Override
    protected void config(String broker, Integer port, Boolean ssl, Boolean withUserNamePass) {

        String protocal = this.TCP;
        if (true == ssl) {
            protocal = this.SSL;
        }

        this.brokerUrl = protocal + this.broker + colon + port;
        this.persistence = new MemoryPersistence();
        this.connectionOptions = new MqttConnectOptions();

        try {
            this.mqttClient = new MqttClient(brokerUrl, clientId, persistence);
            this.connectionOptions.setAutomaticReconnect(true);
            this.connectionOptions.setCleanSession(false);
            this.connectionOptions.setKeepAliveInterval(90);
            this.connectionOptions.setConnectionTimeout(20);

            if (true == withUserNamePass) {
                if (password != null) {
                    this.connectionOptions.setPassword(this.password.toCharArray());
                }
                if (userName != null) {
                    this.connectionOptions.setUserName(this.userName);
                }
            }
            this.mqttClient.connect(this.connectionOptions);
            this.mqttClient.setCallback(this);
        } catch (MqttException me) {
            me.printStackTrace();
        }

    }

    /*
     * (non-Javadoc)
     *
     * @see com.monirthought.config.MQTTConfig#config()
     */
    @Override
    protected void config() {

        this.brokerUrl = this.TCP + this.broker + colon + this.port;
        this.persistence = new MemoryPersistence();
        this.connectionOptions = new MqttConnectOptions();
        try {
            this.mqttClient = new MqttClient(brokerUrl, clientId, persistence);
            this.connectionOptions.setAutomaticReconnect(true);
            this.connectionOptions.setCleanSession(false);
            this.connectionOptions.setKeepAliveInterval(90);
            this.connectionOptions.setConnectionTimeout(20);
            this.connectionOptions.setPassword(this.password.toCharArray());
            this.connectionOptions.setUserName(this.userName);
            this.mqttClient.connect(this.connectionOptions);
            this.mqttClient.setCallback(this);
        } catch (MqttException me) {
            me.printStackTrace();
        }

    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {

        logger.info("MQTTSubcriber client  reconnect is " + reconnect);
        logger.info("MQTTSubcriber client serverURI is " + serverURI);

        this.subscribeMessage(MQTTSubcriberConfig.topic);

    }
}
